package com.alibaba.mqtt.provider.Subscriber.impl;

import com.alibaba.common.api.service.IMQTTSubscriberService;
import com.alibaba.mqtt.provider.config.BaseMqttConfig;
import org.apache.dubbo.config.annotation.Service;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;

@Service(interfaceName="IMQTTSubscriberService", version = "1.0.0",group = "demo")
public class MQTTSubscriberServiceImpl implements IMQTTSubscriberService, MqttCallback {
    private String brokerUrl = null;
    private MqttClient mqttClient = null;
    private MqttConnectOptions connectionOptions = null;
    private MemoryPersistence persistence = null;

    private static final Logger logger = LoggerFactory.getLogger(MQTTSubscriberServiceImpl.class);

    public MQTTSubscriberServiceImpl() {
        this.config();
    }

    /*
     * 连接丢失后，这里面进行重连
     */
    @Override
    public void connectionLost(Throwable cause) {
        logger.info("Connection Lost");

    }

    /*
     * 收到消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        // Called when a message arrives from the server that matches any
        // subscription made by the client
        String time = new Timestamp(System.currentTimeMillis()).toString();
        System.out.println();
        System.out.println("***********************************************************************");
        System.out.println("Message Arrived at Time: " + time + "  Topic: " + topic + "  Message: "
                + new String(message.getPayload(), "GB2312"));
        System.out.println("***********************************************************************");
        System.out.println();
    }

    /*
     *
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        // Leave it blank for subscriber
        logger.info("delivery completed----"  + token.isComplete() );
    }

    /*
     * 订阅主题
     */
    @Override
    public void subscribeMessage(String topic) {
        try {
            this.mqttClient.subscribe(topic, BaseMqttConfig.qos);
        } catch (MqttException me) {
            me.printStackTrace();
        }
    }

    /*
     * 断开连接
     */
    public void disconnect() {
        try {
            this.mqttClient.disconnect();
        } catch (MqttException me) {
            logger.error("ERROR", me);
        }
    }

    /*
     * 自定义配置
     */
    protected void config(String broker, Integer port, Boolean ssl, Boolean withUserNamePass) {

        String protocal = BaseMqttConfig.TCP;
        if (ssl) {
            protocal = BaseMqttConfig.SSL;
        }

        this.brokerUrl = protocal + broker + BaseMqttConfig.colon + port;
        this.persistence = new MemoryPersistence();
        this.connectionOptions = new MqttConnectOptions();

        try {
            this.mqttClient = new MqttClient(brokerUrl, BaseMqttConfig.subClientId, persistence);
            this.connectionOptions.setCleanSession(true);
            if (true == withUserNamePass) {
                if (BaseMqttConfig.password != null) {
                    this.connectionOptions.setPassword(BaseMqttConfig.password.toCharArray());
                }
                if (BaseMqttConfig.userName != null) {
                    this.connectionOptions.setUserName(BaseMqttConfig.userName);
                }
            }
            this.mqttClient.connect(this.connectionOptions);
            this.mqttClient.setCallback(this);
        } catch (MqttException me) {
            me.printStackTrace();
        }

    }

    /*
     *默认配置
     */
    protected void config() {

        this.brokerUrl = BaseMqttConfig.TCP + BaseMqttConfig.broker + BaseMqttConfig.colon + BaseMqttConfig.port;
        this.persistence = new MemoryPersistence();
        this.connectionOptions = new MqttConnectOptions();
        try {
            this.mqttClient = new MqttClient(brokerUrl, BaseMqttConfig.subClientId, persistence);
            this.connectionOptions.setCleanSession(true);
            this.mqttClient.connect(this.connectionOptions);
            this.mqttClient.setCallback(this);
        } catch (MqttException me) {
            me.printStackTrace();
        }

    }

}

